﻿using RabbitMQ.Client;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace RabbitMQExtension
{
    /// <summary>
    /// MQProducerClient
    /// RabbitMQ发布消息客户端。
    /// 
    /// 修改纪录
    /// 
    ///		2020.04.23 版本：1.0 Sunnyfish 创建文件。
    /// 
    /// 版本：1.0
    /// 
    /// <author>
    ///		<name>Sunnyfish</name>
    ///		<date>2020.04.23</date>
    /// </author> 
    /// </summary>
    public class MQProducerClient : IMQProducerClient
    {
        /// <summary>
        /// 日记打印Log，可以外部传入接管默认Log
        /// </summary>
        public ILog Log { get; set; } = LogManager.GetLogger(typeof(MQProducerClient));

        private MQConfig _MQConfig = null;
        private ConnectionFactory _Factory = null;
        private IConnection _Connection = null;
        private IModel _Channel = null;
        private SpinLock _ChannelLock = new SpinLock();
        private object _objLock = new object();

        /// <summary>
        /// 获取消息的发送通道实例。
        /// </summary>
        public IModel MessageChannel
        {
            get
            {
                if (_Channel == null || _Factory == null || _Connection == null)
                {
                    bool isLock = false;
                    _ChannelLock.Enter(ref isLock);
                    if (isLock)
                    {
                        try
                        {
                            if (_Channel == null || _Factory == null || _Connection == null)
                            {
                                ConnectToServer();
                                Log.Info($"Chanel={_Channel},_Factory={_Factory},_Connection={_Connection}");
                            }
                        }
                        catch (Exception e)
                        {
                            Log.Error(e);
                            if (_Connection != null)
                            {
                                _Connection.Close();
                                _Connection.Dispose();
                                _Connection = null;
                            }
                            if (_Channel != null)
                            {
                                _Channel.Close();
                                _Channel.Dispose();
                                _Channel = null;
                            }
                        }
                        finally
                        {
                            _ChannelLock.Exit();
                        }
                    }
                }
                return _Channel;
            }
        }

        /// <summary>
        /// 发送发布/订阅消类别的消息到服务器，同一个交换机的订阅队列的消息都是相同的。
        /// </summary>
        /// <returns><see cref="MQSendResult"/>。</returns>
        public MQSendResult SendPubSubMessage(string topic, string message, string msgId)
        {
            if (string.IsNullOrEmpty(msgId))
                msgId = Guid.NewGuid().ToString();
            bool success = false;
            string exceptionInfo = null;
            try
            {
                SendMessage(msgId, message, topic, "");
                success = true;
            }
            catch (ObjectDisposedException ode)
            {
                Log.Error(ode);
                //如果出现ObjectDisposedException异常，先释放所有资源，然后重新连接。
                if (_Channel != null)
                {
                    if (_Channel.IsOpen)
                        _Channel.Close();
                    _Channel.Dispose();
                    _Channel = null;
                }

                if (_Connection != null)
                {
                    if (_Connection.IsOpen)
                        _Connection.Close();
                    _Connection.Dispose();
                    _Connection = null;
                }

                _Factory = null;
            }
            catch (Exception e)
            {
                Log.Warn($"发送失败，topic={topic}，message={message}，异常={e}");
                Log.Error(e);
                exceptionInfo = e.Message;
            }
            MQSendResult msr = new MQSendResult(msgId, success, exceptionInfo);
            return msr;
        }

        /// <summary>
        /// 发送主题类别的消息到服务器，客户端可以根据多级多题（关键字）过滤进行消息订阅。
        /// </summary>
        /// <param name="topic">消息主题。</param>
        /// <param name="filterExpress">消息过滤关键字，依据此关键字决定消息送往不同的队列。。</param>
        /// <param name="message">消息内容。</param>
        /// <param name="msgId">消息的唯一Id。</param>
        /// <returns><see cref="MQSendResult"/>。</returns>
        public MQSendResult SendTopicMessage(string topic, string filterExpress, string message, string msgId)
        {
            if (string.IsNullOrEmpty(msgId))
                msgId = Guid.NewGuid().ToString();
            bool success = false;
            string exceptionInfo = null;
            try
            {
                var routingKey = string.IsNullOrEmpty(filterExpress) ? topic : filterExpress;
                SendMessage(msgId, message, topic, routingKey);
                success = true;
            }
            catch (Exception e)
            {
                Log.Error(e);
            }
            MQSendResult msr = new MQSendResult(msgId, success, exceptionInfo);
            return msr;
        }

        /// <summary>
        /// 发送消息到指定的交换机。
        /// </summary>
        /// <param name="msgId">消息唯一Id。</param>
        /// <param name="message">消息内容。</param>
        /// <param name="exchangeName">消息交换机名称。</param>
        /// <param name="routingKey">消息路由关键字。</param>
        private void SendMessage(string msgId, string message, string exchangeName, string routingKey)
        {
            IBasicProperties bp = MessageChannel.CreateBasicProperties();
            bp.MessageId = msgId;
            // 队列数据 数据持久化 
            bp.DeliveryMode = 2;

            var body = Encoding.UTF8.GetBytes(message);
            try
            {
                MessageChannel.BasicPublish(exchange: exchangeName,
                                         routingKey: routingKey,
                                         basicProperties: bp,
                                         body: body);
            }
            catch (AlreadyClosedException ace)
            {
                //连接已关闭，重置连接。
                _Factory = null;
                throw ace;
            }
        }

        /// <summary> 保存MQ连接字典 </summary>
        private static ConcurrentDictionary<string, IModel> _modelDic = new ConcurrentDictionary<string, IModel>();

        /// <summary>
        /// 发送 消息 到队列，直接发送模式，队列不存在 不会创建队列，不会异常
        /// </summary>
        /// <param name="queue">队列名</param>
        /// <param name="message">消息内容</param>
        /// <param name="msgId">消息唯一Id</param>
        /// <returns><see cref="MQSendResult"/>。</returns>
        public MQSendResult SendDirectMessage(string queue, string message, string msgId)
        {
            if (string.IsNullOrEmpty(queue))
            {
                throw new ArgumentNullException(nameof(queue));
            }
            if (string.IsNullOrEmpty(message))
            {
                throw new ArgumentNullException(nameof(message));
            }
            if (string.IsNullOrEmpty(msgId))
                msgId = Guid.NewGuid().ToString();
            bool sucess = false;
            string exceptionInfo = null;
            try
            {
                lock (_objLock)
                    _Channel.BasicPublish("", queue, null, Encoding.UTF8.GetBytes(message));
                sucess = true;
            }
            catch (Exception e)
            {
                exceptionInfo = e.Message;
                Log.Error(e);
            }

            return new MQSendResult(msgId, sucess, exceptionInfo);
        }

        /// <summary>
        /// 设置消息队列的连接信息。
        /// </summary>
        /// <param name="config">信息实体</param>
        public void SetConnectInfo(MQConfig config)
        {
            _MQConfig = config;
            if (MessageChannel != null)
            {
                if (MessageChannel.IsOpen)
                    Log.Info($"消息队列服务器 {config.Host} 连接成功。");
            }
        }

        /// <summary>
        /// 连接到服务器。
        /// </summary>
        private void ConnectToServer()
        {
            _Factory = new ConnectionFactory()
            {
                VirtualHost = string.IsNullOrEmpty(_MQConfig.VirtualHost) ? "/" : _MQConfig.VirtualHost,
                UserName = _MQConfig.UserName,
                Password = _MQConfig.Password,
                NetworkRecoveryInterval = new TimeSpan(0, 0, 5),   //每5秒钟检查一次网络连接。
                AutomaticRecoveryEnabled = true,                    //网络连接异常时自动恢复。
                RequestedHeartbeat = 30                             //心跳检测超时时间，30秒。
            };
            if (_Connection != null)
            {
                if (_Connection.IsOpen)
                    _Connection.Close();
                _Connection.Dispose();
            }

            if (_Channel != null)
            {
                if (_Channel.IsOpen)
                    _Channel.Close();
                _Channel.Dispose();
            }
            var amqps = new List<AmqpTcpEndpoint>();
            var items = _MQConfig.Host.Split(new char[] { ',', ';' }, StringSplitOptions.RemoveEmptyEntries);
            foreach (var item in items)
            {
                var hosts = item.Split(':');
                var port = 5672;
                if (hosts.Length > 1)
                    int.TryParse(hosts[1], out port);
                if (port < 1)
                    port = 5672;
                amqps.Add(new AmqpTcpEndpoint
                {
                    HostName = hosts[0],
                    Port = port,
                });
            }
            _Connection = _Factory.CreateConnection(amqps);
            _Channel = _Connection.CreateModel();
        }

        /// <summary>
        /// 与服务器是否处于连接状态。
        /// </summary>
        /// <returns>true:已连接，false:未连接。</returns>
        public bool IsConnected()
        {
            bool isConnect = false;
            if (_Connection != null)
                isConnect = _Connection.IsOpen;
            return isConnect;
        }

        /// <summary>
        /// 关闭客户端。
        /// </summary>
        public void Shutdown()
        {
            if (_Channel != null)
            {
                if (_Channel.IsOpen)
                    _Channel.Close();

            }

            if (_Connection != null)
            {
                if (_Connection.IsOpen)
                    _Connection.Close();
            }
        }

        public void Dispose()
        {
            Shutdown();
            try
            {
                if (_Channel != null)
                {
                    if (_Channel.IsOpen)
                        _Channel.Close();
                    _Channel.Dispose();
                }
                if (_Connection != null)
                {
                    if (_Connection.IsOpen)
                        _Connection.Close();
                    _Connection.Dispose();
                }
            }
            finally
            {
                _Factory = null;
                _Channel = null;
                _Connection = null;
            }
        }
    }
}
